//! 内存存储实现
//!
//! 本模块提供了基于内存的高性能键值存储实现。
//! 使用内存作为主要存储介质，适用于对性能要求极高的场景：
//!
//! - **超高性能**: 纯内存操作，无磁盘IO延迟
//! - **线程安全**: 使用DashMap和读写锁保证并发安全
//! - **事务支持**: 支持内存事务操作
//! - **零拷贝**: 尽可能减少数据复制操作
//! - **实时性**: 数据操作立即生效，无延迟
//!
//! ## 内存结构
//!
//! 数据存储在以下结构中：
//! - `DashMap<String, Vec<u8>>`: 高性能并发哈希表
//! - `Arc<RwLock<...>>`: 线程安全的引用计数包装
//!
//! ## 适用场景
//!
//! - **缓存系统**: 作为高速缓存层
//! - **临时数据**: 会话数据、临时计算结果
//! - **测试环境**: 单元测试、集成测试
//! - **开发调试**: 快速验证功能
//!
//! ## 性能特点
//!
//! - **O(1) 读写**: 哈希表实现，常数时间复杂度
//! - **并发友好**: 支持多线程并发读写
//! - **零GC压力**: 使用Rust的所有权系统，无垃圾回收
//!
//! ## 使用示例
//!
//! ```rust
//! use nacos_storage::memory::MemoryStorage;
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
//!     // 创建内存存储实例
//!     let storage = MemoryStorage::new();
//!     
//!     // 存储数据
//!     storage.put("service:user", b"user data").await?;
//!     
//!     // 读取数据
//!     let data = storage.get("service:user").await?;
//!     assert_eq!(data, Some(b"user data".to_vec()));
//!     
//!     // 检查大小
//!     assert_eq!(storage.size(), 1);
//!     
//!     Ok(())
//! }
//! ```
//!
//! ## 事务支持
//!
//! 支持内存事务操作，提供ACID特性：
//! - **原子性**: 事务内操作要么全部成功，要么全部失败
//! - **一致性**: 事务执行前后数据保持一致状态
//! - **隔离性**: 并发事务之间互不干扰
//! - **持久性**: 内存数据在事务提交后立即生效
//!
//! ## 注意事项
//!
//! - **非持久化**: 服务重启后数据会丢失
//! - **内存限制**: 受系统内存大小限制
//! - **冷启动**: 每次启动后需要重新加载数据
//!
//! ## 性能对比
//!
//! | 操作类型 | 内存存储 | 文件存储 | SQLite |
//! |----------|----------|----------|--------|
//! | 写入延迟 | ~1μs     | ~1ms     | ~100μs |
//! | 读取延迟 | ~100ns   | ~500μs   | ~10μs  |
//! | 并发能力 | 高        | 中       | 中     |
//! | 持久化   | 否        | 是       | 是     |

use async_trait::async_trait;
use dashmap::DashMap;
use nacos_core::{Storage, Result};
use parking_lot::{RwLock};
use std::collections::HashMap;
use std::sync::Arc;
use tracing::{debug, trace};
use crate::transaction::{TransactionalStorage, MemoryTransaction};

/// 内存存储实现
///
/// `MemoryStorage` 提供了基于内存的高性能键值存储功能。
/// 使用DashMap实现线程安全的并发哈希表，支持高并发读写操作。
#[derive(Debug, Clone)]
pub struct MemoryStorage {
    /// 底层数据存储
    ///
    /// 使用DashMap实现并发安全的哈希表：
    /// - `DashMap`: 支持分片锁的并发哈希表，减少锁竞争
    /// - `Arc`: 线程安全的引用计数，支持多实例共享
    /// - `RwLock`: 读写分离锁，支持并发读
    pub(crate) data: Arc<RwLock<DashMap<String, Vec<u8>>>>,
}

impl MemoryStorage {
    /// 创建新的内存存储实例
    ///
    /// # 返回值
    /// - `MemoryStorage`: 新创建的内存存储实例
    ///
    /// # 行为
    /// 初始化空的DashMap，准备好接收数据。
    /// 该操作是O(1)时间复杂度。
    pub fn new() -> Self {
        trace!("Creating new MemoryStorage instance");
        Self {
            data: Arc::new(RwLock::new(DashMap::new())),
        }
    }

    /// 获取当前存储的键值对数量
    ///
    /// # 返回值
    /// - `usize`: 当前存储的键值对数量
    ///
    /// # 行为
    /// 获取当前内存中存储的总记录数。
    /// 这是一个O(1)时间复杂度的操作。
    pub fn size(&self) -> usize {
        self.data.read().len()
    }
}

impl Default for MemoryStorage {
    /// 创建默认的内存存储实例
    ///
    /// # 返回值
    /// - `MemoryStorage`: 默认创建的内存存储实例
    ///
    /// # 行为
    /// 与`new()`方法相同，创建新的空存储实例。
    /// 实现了Default trait，方便在结构体中使用。
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl Storage for MemoryStorage {
    /// 根据键获取对应的值
    ///
    /// # 参数
    /// - `key`: 要获取的键字符串
    ///
    /// # 返回值
    /// - `Ok(Some(Vec<u8>))`: 成功获取到数据
    /// - `Ok(None)`: 键不存在
    /// - `Err(_)`: 理论上不会失败，因为内存操作不会出错
    ///
    /// # 性能
    /// O(1)时间复杂度，使用哈希表查找。
    /// 使用读锁保证并发安全，支持多线程并发读取。
    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
        trace!("Getting key: {}", key);
        let lock = self.data.read();
        let result = lock.get(key).map(|v| v.clone());
        debug!("Get result for key {}: {:?}", key, result.is_some());
        Ok(result)
    }

    /// 存储键值对
    ///
    /// # 参数
    /// - `key`: 键字符串
    /// - `value`: 要存储的二进制数据
    ///
    /// # 返回值
    /// - `Ok(())`: 存储成功
    /// - `Err(_)`: 理论上不会失败
    ///
    /// # 性能
    /// O(1)时间复杂度，哈希表插入操作。
    /// 使用写锁保证并发安全，同一时间只允许一个写入操作。
    ///
    /// # 原子性
    /// 单个键的写入操作是原子的。
    async fn put(&self, key: &str, value: &[u8]) -> Result<()> {
        trace!("Putting key: {}, value size: {}", key, value.len());
        let lock = self.data.write();
        lock.insert(key.to_string(), value.to_vec());
        debug!("Put completed for key: {}", key);
        Ok(())
    }

    /// 删除指定键对应的数据
    ///
    /// # 参数
    /// - `key`: 要删除的键字符串
    ///
    /// # 返回值
    /// - `Ok(())`: 删除成功（包括键不存在的情况）
    /// - `Err(_)`: 理论上不会失败
    ///
    /// # 性能
    /// O(1)时间复杂度，哈希表删除操作。
    /// 使用写锁保证并发安全。
    ///
    /// # 幂等性
    /// 删除不存在的键会返回Ok(())，不会报错。
    async fn delete(&self, key: &str) -> Result<()> {
        trace!("Deleting key: {}", key);
        let lock = self.data.write();
        let existed = lock.remove(key).is_some();
        debug!("Delete result for key {}: existed={}", key, existed);
        Ok(())
    }

    /// 扫描指定前缀的所有键值对
    ///
    /// # 参数
    /// - `prefix`: 键前缀字符串
    ///
    /// # 返回值
    /// - `Ok(Vec<(String, Vec<u8>)>>`: 匹配前缀的所有键值对
    /// - `Err(_)`: 理论上不会失败
    ///
    /// # 性能
    /// O(n)时间复杂度，需要遍历所有键值对。
    /// n是存储中的总键数。
    ///
    /// # 行为
    /// 遍历内存中的所有键值对，返回以指定前缀开头的所有数据。
    async fn scan(&self, prefix: &str) -> Result<Vec<(String, Vec<u8>)>> {
        let lock = self.data.read();
        let result = lock.iter()
            .filter(|entry| entry.key().starts_with(prefix))
            .map(|entry| (entry.key().clone(), entry.value().clone()))
            .collect();
        Ok(result)
    }
    
    /// 批量获取多个键对应的值
    ///
    /// # 参数
    /// - `keys`: 要获取的键列表
    ///
    /// # 返回值
    /// - `Ok(HashMap<String, Vec<u8>>)`: 成功获取的键值对映射
    ///   - 只包含实际存在的键值对
    ///   - 不存在的键会被忽略
    /// - `Err(_)`: 理论上不会失败
    ///
    /// # 性能
    /// O(m)时间复杂度，m是要获取的键的数量。
    /// 每个键的获取都是O(1)操作。
    ///
    /// # 并发性
    /// 使用读锁，支持与其他读取操作并发执行。
    async fn batch_get(&self, keys: &[String]) -> Result<HashMap<String, Vec<u8>>> {
        let lock = self.data.read();
        let mut result = HashMap::new();
        for key in keys {
            if let Some(value) = lock.get(key) {
                result.insert(key.clone(), value.clone());
            }
        }
        Ok(result)
    }

    /// 批量存储多个键值对
    ///
    /// # 参数
    /// - `data`: 要存储的键值对映射
    ///
    /// # 返回值
    /// - `Ok(())`: 批量存储成功
    /// - `Err(_)`: 理论上不会失败
    ///
    /// # 性能
    /// O(m)时间复杂度，m是要存储的键值对数量。
    /// 每个键值对的存储都是O(1)操作。
    ///
    /// # 原子性
    /// 这不是原子操作。如果中途遇到错误，部分数据可能已成功存储。
    /// 但因为内存操作不会失败，所以实际上会全部成功。
    async fn batch_put(&self, data: &HashMap<String, Vec<u8>>) -> Result<()> {
        let lock = self.data.write();
        for (key, value) in data {
            lock.insert(key.clone(), value.clone());
        }
        Ok(())
    }

    /// 批量删除多个键对应的数据
    ///
    /// # 参数
    /// - `keys`: 要删除的键列表
    ///
    /// # 返回值
    /// - `Ok(())`: 批量删除成功
    /// - `Err(_)`: 理论上不会失败
    ///
    /// # 性能
    /// O(m)时间复杂度，m是要删除的键的数量。
    /// 每个键的删除都是O(1)操作。
    ///
    /// # 幂等性
    /// 删除不存在的键会被忽略，不会报错。
    async fn batch_delete(&self, keys: &[String]) -> Result<()> {
        let lock = self.data.write();
        for key in keys {
            lock.remove(key);
        }
        Ok(())
    }

    /// 检查指定键是否存在
    ///
    /// # 参数
    /// - `key`: 要检查的键字符串
    ///
    /// # 返回值
    /// - `Ok(true)`: 键存在
    /// - `Ok(false)`: 键不存在
    /// - `Err(_)`: 理论上不会失败
    ///
    /// # 性能
    /// O(1)时间复杂度，哈希表查找操作。
    /// 使用读锁保证并发安全。
    async fn exists(&self, key: &str) -> Result<bool> {
        Ok(self.data.read().contains_key(key))
    }

    /// 获取匹配指定模式的键列表
    ///
    /// # 参数
    /// - `pattern`: 模式字符串，支持通配符
    ///   - `*`: 匹配任意字符串（如"user:*"匹配所有以"user:"开头的键）
    ///   - `*suffix`: 匹配以指定后缀结尾的键
    ///   - 精确匹配：匹配完全相等的键
    ///
    /// # 返回值
    /// - `Ok(Vec<String>)`: 匹配的键列表
    /// - `Err(_)`: 理论上不会失败
    ///
    /// # 性能
    /// O(n)时间复杂度，需要遍历所有键。
    /// n是存储中的总键数。
    ///
    /// # 模式匹配规则
    /// - "*"：匹配所有键
    /// - "prefix*"：匹配以指定前缀开头的键
    /// - "*suffix"：匹配以指定后缀结尾的键
    /// - "exact"：精确匹配
    async fn keys(&self, pattern: &str) -> Result<Vec<String>> {
        let lock = self.data.read();
        let result = lock.iter()
            .map(|entry| entry.key().clone())
            .filter(|k| self.match_pattern(k, pattern))
            .collect();
        Ok(result)
    }

    /// 清空所有存储的数据
    ///
    /// # 返回值
    /// - `Ok(())`: 清空成功
    /// - `Err(_)`: 理论上不会失败
    ///
    /// # 行为
    /// 立即清空内存中的所有数据，释放相关内存。
    /// 这是一个破坏性操作，无法恢复。
    ///
    /// # 性能
    /// O(1)时间复杂度，直接清空哈希表。
    ///
    /// # 警告
    /// 这是一个破坏性操作，会永久删除所有存储的数据。
    /// 因为内存存储是非持久化的，清空后数据无法恢复。
    async fn clear(&self) -> Result<()> {
        self.data.write().clear();
        Ok(())
    }
}

impl MemoryStorage {
    /// 模式匹配函数
    ///
    /// # 参数
    /// - `key`: 要匹配的键字符串
    /// - `pattern`: 模式字符串
    ///
    /// # 返回值
    /// - `true`: 键匹配模式
    /// - `false`: 键不匹配模式
    ///
    /// # 支持的模式
    /// - "*"：匹配任意字符串
    /// - "prefix*"：匹配以指定前缀开头的字符串
    /// - "*suffix"：匹配以指定后缀结尾的字符串
    /// - 精确匹配：完全相等的字符串
    ///
    /// # 示例
    /// ```
    /// assert!(storage.match_pattern("user:123", "user:*"));
    /// assert!(storage.match_pattern("user:123", "*123"));
    /// assert!(storage.match_pattern("user:123", "user:123"));
    /// assert!(!storage.match_pattern("user:123", "admin:*"));
    /// ```
    fn match_pattern(&self, key: &str, pattern: &str) -> bool {
        if pattern == "*" { return true; }
        if let Some(prefix) = pattern.strip_suffix('*') {
            return key.starts_with(prefix);
        }
        if let Some(suffix) = pattern.strip_prefix('*') {
            return key.ends_with(suffix);
        }
        key == pattern
    }
}

#[async_trait]
impl TransactionalStorage for MemoryStorage {
    /// 事务类型定义
    ///
    /// 使用MemoryTransaction作为内存存储的事务类型。
    type Tx = MemoryTransaction;

    /// 开始一个新的事务
    ///
    /// # 返回值
    /// - `Ok(MemoryTransaction)`: 成功创建的事务对象
    /// - `Err(_)`: 理论上不会失败
    ///
    /// # 行为
    /// 创建新的内存事务实例，该事务将操作当前的内存存储。
    ///
    /// # 事务特性
    /// 内存事务提供以下特性：
    /// - 原子性：事务内操作要么全部成功，要么全部失败
    /// - 一致性：保证数据状态的一致性
    /// - 隔离性：事务之间相互隔离
    ///
    /// # 使用示例
    /// ```rust
    /// let storage = MemoryStorage::new();
    /// let mut tx = storage.begin_transaction().await.unwrap();
    /// // 在事务中执行操作...
    /// tx.commit().await.unwrap();
    /// ```
    async fn begin_transaction(&self) -> Result<Self::Tx> {
        trace!("Beginning memory transaction");
        Ok(MemoryTransaction::new(self.clone()))
    }
}


#[cfg(test)]
mod tests {
    use super::*;
    use crate::transaction::Transaction;

    #[tokio::test]
    async fn test_put_and_get() {
        let storage = MemoryStorage::new();
        storage.put("key", b"value").await.unwrap();
        assert_eq!(storage.get("key").await.unwrap(), Some(b"value".to_vec()));
    }

    #[tokio::test]
    async fn test_transaction_commit() {
        let storage = MemoryStorage::new();
        let mut tx = storage.begin_transaction().await.unwrap();
        
        // In a real scenario, the transaction object would have methods to operate on data.
        // In this simplified model, we operate on the storage directly,
        // and the transaction just controls the "commit" (which is a no-op here).
        tx.put("tx_key", b"tx_value").await.unwrap();

        tx.commit().await.unwrap();

        assert_eq!(storage.get("tx_key").await.unwrap(), Some(b"tx_value".to_vec()));
    }
}

#[async_trait::async_trait]
impl nacos_core::EventPublisher for MemoryStorage {
    async fn publish(&self, _event: nacos_core::Event) -> nacos_core::Result<()> {
        Ok(())
    }
    
    async fn subscribe(&self, _listener: Box<dyn nacos_core::EventListener>) -> nacos_core::Result<()> {
        Ok(())
    }
    
    async fn unsubscribe(&self, _listener_name: &str) -> nacos_core::Result<()> {
        Ok(())
    }
}